package com.yusc.utils;

import com.google.gson.Gson;
import com.yusc.pojo.ALlBasicInfo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.List;

/**
 * @author lglyoung
 */
public class WriteHdfsHandler {
    /**
     * 父路径，即数据文件所存放的路径
     */
    private String canDataDir;

    /**
     * 文件名
     */
    private final String FILE_NAME = "air_data.json.";

    /**
     * 文件名后缀
     */
    private int curFileNameSuffix = 1;

    /**
     * 写了多少行数据到当前文件
     */
    private int curLineNum = 0;

    /**
     * 每个文件最大行数，超过，则写入下一个文件
     */
    private final int MAX_LINE_NUMS = 1000000;

    private FileSystem fs;

    /**
     * 构造函数
     * @param canDataDir
     */
    public WriteHdfsHandler(String canDataDir, String defaultFS) throws IOException {
        this.canDataDir = canDataDir;

        // 获取HDFS文件系统对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", defaultFS);
        fs = FileSystem.get(conf);

        // 父目录删除
        boolean exists = fs.exists(new Path(this.canDataDir));
        if (exists) {
            fs.delete(new Path(this.canDataDir), true);
        }
        fs.mkdirs(new Path(this.canDataDir));
    }

    /**
     * 获取当前can data文件路径
     * @return
     */
    public String getCurCanDataFilePath() {
        return this.canDataDir + "/" + FILE_NAME + this.curFileNameSuffix;
    }

    /**
     * 将车辆数据写入到HDFS
     * @param infoList
     */
    public void writeCanDataToHdfs(List<ALlBasicInfo> infoList) throws IOException {
        if(infoList.isEmpty()) {
            return ;
        }

        int curHandedIndex = 0;
        // 循环写入新的文件
        while(curHandedIndex < infoList.size()) {
            if (curLineNum >= MAX_LINE_NUMS) {
                this.curFileNameSuffix += 1;
                curLineNum = 0;
            }

            // 创建新的文件
            String curCanDataFilePath = getCurCanDataFilePath();
            FSDataOutputStream out = null;
            if(fs.exists(new Path(curCanDataFilePath))) {
                out  = fs.append(new Path(curCanDataFilePath));
            } else {
                out = fs.create(new Path(curCanDataFilePath));
            }
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
            Gson gson = new Gson();

            // 循环将数据写入HDFS
            for(; curHandedIndex < infoList.size() && curLineNum < MAX_LINE_NUMS; curHandedIndex++) {
                ALlBasicInfo basicInfo = infoList.get(curHandedIndex);
                writer.write(gson.toJson(basicInfo)+"\r\n");
                curLineNum ++;
            }
            writer.close();
        }
        // 特殊处理：避免追加操作。伪分布式下，append会报错。
        if(this.curLineNum > 0 && this.curLineNum < MAX_LINE_NUMS) {
            this.curFileNameSuffix ++;
            curLineNum = 0;
        }
    }

    public void close() throws IOException {
        fs.close();
    }
}
